package com.yulp.springboot2task.config;

import com.alibaba.fastjson.JSON;
import com.yulp.springboot2task.service.TestJesQueImpl;
import com.yulp.springboot2task.test.PaymentConsumerRetryHandler;
import net.greghaines.jesque.Config;
import net.greghaines.jesque.ConfigBuilder;
import net.greghaines.jesque.client.ClientPoolImpl;
import net.greghaines.jesque.utils.PoolUtils;
import net.greghaines.jesque.worker.*;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.List;

import static net.greghaines.jesque.utils.JesqueUtils.entry;
import static net.greghaines.jesque.utils.JesqueUtils.map;

/**
 * jesque执行流程
 * 提交任务：使用Client提交到redis，需要指定处理job的classname与队列名。
 * 获取任务：WorkerImpl类的poll方法，从redis中获取任务： 先获取队列名，再根据队列名获取值。
 * 执行任务：从redis获取任务后，将任务转成job，得到任务的实际执行类，再调用任务的run方法。
 *
 * jesque组成：1个config, 1个constant, 1个发送类Sender，n个消费类Consumer
 */

@Component
public class JesqueConfig {

    @Value("${spring.redis.host}")
    private String host;

    private int port = 6379;

    @Value("${spring.redis.password}")
    private String password;

    public JesqueConfig() {
    }

    public Config config() {
        ConfigBuilder configBuilder = new ConfigBuilder().withHost(host).withPort(port);
        if (StringUtils.isNotBlank(password)) {
            configBuilder.withPassword(password);
        }
        return configBuilder.build();
    }

    @Bean
    public ClientPoolImpl jesqueClientPool() {
        return new ClientPoolImpl(config(), PoolUtils.createJedisPool(config()));
    }

    @Bean
    public RetryWorkMessageConsumer RetryWorkMessageConsumer() {
        return new RetryWorkMessageConsumer();
    }

    /**
     * 这里可以提出来，做成消费者Consumer
     */
    @Component
    public class RetryWorkMessageConsumer {

        private Logger logger = LoggerFactory.getLogger(this.getClass());

        @Autowired
        private JesqueConfig jesqueConfig;

        private Worker worker;

        private Thread workerThread;

        // 启动监听队列。
        @PostConstruct
        public void init() {

            List list = new ArrayList<String>() {{
                add(TestJesQueImpl.PAYMET_RETRY_QUEUE_NAME); //使用枚举：RetryMessageEnum.soPay2cn.getQueueName()
               /* add(TestJesQueImpl.REFUND_RETRY_QUEUE_NAME);*/
            }};

            JobFactory jobFactory = new MapBasedJobFactory(map(
                    entry(TestJesQueImpl.PAYMET_RETRY_ACTION_NAME, PaymentConsumerRetryHandler.class)/*,
                    entry(TestJesQueImpl.PAYMET_RETRY_ACTION_NAME, TestAction.class)*/));
            //使用枚举：RetryMessageEnum.soCancle2cn.getClassName(), SoCancel2CnRetryConsumer.class

            worker = new WorkerImpl(jesqueConfig.config(), list, jobFactory);

            worker.getWorkerEventEmitter().addListener((event, worker, queue, job, runner, result, t) -> {
                logger.info("JOB_EXECUTE");
            }, WorkerEvent.JOB_EXECUTE);

            worker.getWorkerEventEmitter().addListener((event, worker, queue, job, runner, result, t) -> {
                logger.info("JOB_SUCCESS");
            }, WorkerEvent.JOB_SUCCESS);

            worker.getWorkerEventEmitter().addListener((event, worker, queue, job, runner, result, t) -> {
                logger.error("JOB_FAILURE {}-{}", JSON.toJSONString(job), t);
            }, WorkerEvent.JOB_FAILURE);

            workerThread = new Thread(worker);
            workerThread.start();
        }

        @PreDestroy
        public void destroy() {
            worker.end(true);
            try {
                workerThread.join();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

}
